package com.zhuliwei.rabbitmq;

import com.zhuliwei.constants.RabbitConstants;
import com.zhuliwei.utils.RedisCache;
import jakarta.annotation.PostConstruct;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.ReturnedMessage;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Objects;

/**
 * 实现发布确认回调接口RabbitTemplate.ConfirmCallback(是一个内部接口)
 *
 * @author 朱力维
 * @time 2024/3/8 18:03
 **/
@Component
@Slf4j
public class MyCallBack implements RabbitTemplate.ConfirmCallback, RabbitTemplate.ReturnsCallback {

    @Resource
    private RedisCache redisCache;

    @Resource
    private RabbitTemplate rabbitTemplate;

    //依赖注入完成之后需要执行的方法
    @PostConstruct
    public void init() {
        rabbitTemplate.setConfirmCallback(this::confirm);
        rabbitTemplate.setReturnsCallback(this::returnedMessage);
    }

    /**
     * /**
     * 交换机不管是否收到消息的一个回调方法
     *
     * @param correlationData 消息相关数据
     * @param ack             交换机是否收到消息
     * @param cause           为收到消息的原因
     */
    @Override
    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
        if (Objects.isNull(correlationData))
            return;
        String id = correlationData.getId() == null ? "" : correlationData.getId();
        if (ack) {
            // 当交换机收到消息从缓存中删除已收到的消息
            redisCache.delCacheMapValue(RabbitConstants.RABBIT_RENEW, id);
        } else {
            // 如果消息投递到交互机失败打印错误信息
            log.error("交换机还未收到 id 为:{}消息，原因:{}", id, cause);
        }
    }

    //当消息未路由到队列时触发
    @Override
    public void returnedMessage(ReturnedMessage returned) {
        try {
            // 每10s重发一次消息
            Thread.sleep(10000);
            // 需重发消息(邮箱类)的使用消息回退，无需重发消息只需备份和报警（chatgpt类）使用备份交换机，备份交换机优先级高于消息回退
            String messageId = returned.getMessage().getMessageProperties().getMessageId() == null ? "" : returned.getMessage().getMessageProperties().getMessageId();
            // 消息退回并重发
            if (!StringUtils.hasText(messageId)) {
                log.error("消息:{}无法路由到队列退回并重发，退回原因:{}",new String(returned.getMessage().getBody()),returned.getReplyText());
                rabbitTemplate.convertAndSend(returned.getExchange(), returned.getRoutingKey(), new String(returned.getMessage().getBody()));
            } else {
                log.error("消息:{}无法路由到队列退回并重发，退回原因:{}",new String(returned.getMessage().getBody()),returned.getReplyText());
                rabbitTemplate.convertAndSend(returned.getExchange(), returned.getRoutingKey(), new String(returned.getMessage().getBody()), new CorrelationData(messageId));
            }
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
